// Copyright 2012 Google Inc. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include "connection.h"

#include <deque>
#include <sstream>

#include "event2/buffer.h"
#include "event2/bufferevent.h"
#include "event2/event.h"
#include "event2/thread.h"
#include "glog/logging.h"

#include "connection_queue.h"

namespace seven_summits {

#define MAX_REQUEST_BYTES 1 << 19  // roughly 500k
#define MAX_RESPONSE_BYTES 1 << 19  // roughly 500k

Connection::BufferEvent::BufferEvent(struct event_base* base,
                                     int sfd,
                                     Connection* connection)
    : base_(base),
      sfd_(sfd),
      connection_(connection),
      bev_(bufferevent_socket_new(base_, sfd_,
                                  BEV_OPT_CLOSE_ON_FREE
                                  | BEV_OPT_THREADSAFE)) {
  bufferevent_setcb(bev_, handle_read, NULL, handle_error, connection_);
  bufferevent_enable(bev_, EV_READ);
}

Connection::BufferEvent::~BufferEvent() {
  if (bev_) {
    bufferevent_free(bev_);
  }
}

bool Connection::BufferEvent::ReceivedMoreThan(size_t expected) {
  size_t received = evbuffer_get_length(bufferevent_get_input(bev_));
  return expected <= received;
}

size_t Connection::BufferEvent::Read(void* dest, size_t num_bytes) {
  return bufferevent_read(bev_, dest, num_bytes);
}

void Connection::BufferEvent::Write(const void* dest, size_t num_bytes) {
  if (num_bytes > MAX_RESPONSE_BYTES) {
    num_bytes = MAX_RESPONSE_BYTES;
    LOG(ERROR) << "reponse too large - truncating";
  }
  // The following is x64 clean for suitable MAX_RESPONSE_BYTES.
  uint32_t num_response_bytes = (uint32_t) num_bytes;

  bufferevent_write(bev_, &num_response_bytes, sizeof(uint32_t));
  bufferevent_write(bev_, dest, num_response_bytes);
}

uint32_t Connection::kUndefined = -1;

Connection::Connection()
    : state_(INITIAL),
      callback_(NULL),
      num_request_bytes_(kUndefined),
      smt_request_(),
      smt_response_() {
}

Connection::Connection(BufferEvent* buffer_event)
    : state_(INITIAL),
      callback_(NULL),
      buffer_event_(buffer_event),
      num_request_bytes_(kUndefined),
      smt_request_(),
      smt_response_() {
}

Connection::~Connection() {
  delete buffer_event_;
}

void Connection::Init(struct event_base* base, int fd,
                      ProcessingCallback* callback) {
  buffer_event_ = new BufferEvent(base, fd, this);
  callback_ = callback;
}

void Connection::handle_read(struct bufferevent *bev, void *ptr) {
  Connection* c = reinterpret_cast<Connection*>(ptr);
  c->HandleRead();
}

void Connection::Close() {
  delete buffer_event_;
  if (pool_) {
    pool_->Push(this);
  }
  state_ = INITIAL;
}

void Connection::HandleRead() {
  switch (state_) {
    case INITIAL:
      if (!buffer_event_->ReceivedMoreThan(sizeof(uint32_t))) {
        return;
      }
      buffer_event_->Read(&num_request_bytes_, sizeof(uint32_t));
      VLOG(2) << "received size " << num_request_bytes_;

      if (0 == num_request_bytes_) {
        Close();
        return;
      }
      state_ = READ_MSG;
      // fall-through
 
    case READ_MSG:
      if (!buffer_event_->ReceivedMoreThan(num_request_bytes_)) {
        return;
      }
      ParseRequest();
      VLOG(2) << "received request " << smt_request_.DebugString();;
      if (callback_) {
        callback_->ProcessAsync(smt_request_, &smt_response_, this);
      }
      state_ = WRITE_MSG;
      break;

    default:
      // ignore - will pick this up after writing the response.
      break;
  }
}

void Connection::handle_error(struct bufferevent *bev, short error, void *ptr) {
  VLOG(0) << "ERROR ";
  if (error & BEV_EVENT_EOF) {
    VLOG(0) << "BEV_EVENT_EOF";
    Connection* c = reinterpret_cast<Connection*>(ptr);
    c->Close();
    // connection has been closed, do any clean up here
  } else if (error & BEV_EVENT_ERROR) {
    VLOG(0) << "BEV_EVENT_ERROR";
    // check errno to see what error occurred
    VLOG(0) << "errno : " << errno;
  } else if (error & BEV_EVENT_TIMEOUT) {
    VLOG(0) << "BEV_EVENT_TIMEOUT";
    // must be a timeout event handle, handle it
  }
 
  if (!ptr) {
    return;
  }
  Connection* c = reinterpret_cast<Connection*>(ptr);
  c->HandleError(error);
}

void Connection::HandleError(short error) {
  if (error & BEV_EVENT_EOF) {
    // connection has been closed, do any clean up here
  } else if (error & BEV_EVENT_ERROR) {
    // check errno to see what error occurred
  } else if (error & BEV_EVENT_TIMEOUT) {
    // must be a timeout event handle, handle it
  }
}

bool Connection::ParseRequest() {
  // TODO(bqe): avoid this copy -- ZeroCopyInputStream
  char* buf = (char*) calloc(num_request_bytes_ + 1, sizeof(char));
  if (buf == NULL) {
    LOG(ERROR) << "no memory?";
    return false;
  }
  if (num_request_bytes_ != buffer_event_->Read(buf, num_request_bytes_)) {
    LOG(ERROR) << "couldn't get the data?";
    return false;
  }
  buf[num_request_bytes_] = '\0';
  bool result = smt_request_.ParseFromString(buf);
  free(buf);
  return result;
}

void Connection::WriteResponse() {
  VLOG(2) << "WriteResponse()";
  VLOG(2) << smt_response_.DebugString();
  std::stringstream ss(std::stringstream::in | std::stringstream::out);
  smt_response_.SerializeToOstream(&ss);
  smt_response_.Clear();

  buffer_event_->Write(ss.str().c_str(), ss.str().length());
  state_ = INITIAL;
  if (buffer_event_->ReceivedMoreThan(0)) {
    HandleRead();
  }
}

}  // namespace seven_summits
